package com.kingsoft.dc.khaos.util

import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.{BlockingQueue, LinkedBlockingDeque}

import scala.util.control.NonFatal

/**
  * Created by chengguangqing on 2019/5/27.
  */
private[khaos] abstract class EventLoop[E](name:String) extends Logging{

  private val eventQueue: BlockingQueue[E] = new LinkedBlockingDeque[E]()

  private val stopped = new AtomicBoolean(false)

    private val eventThread = new Thread(name){
    setDaemon(true)

    override def run(): Unit = {
      try{
        while(!stopped.get()){
          val event = eventQueue.take()
          try{
            onReceive(event)
          } catch {
            case  NonFatal(e) => log.error("Unexpected error in "+name,e)
          }
        }
      } catch {
        case ie: InterruptedException =>
        case NonFatal(e) => log.error("Unexpected error in "+name,e)
      }
    }
  }
  protected def onReceive(event: E): Unit

  def start():Unit = {
    if(stopped.get()){
        throw new IllegalArgumentException(s"${name} has already been stoped")
    }

    onStart()
    log.info("开始运行处理线程")
    eventThread.start()
  }

  def stop():Unit = {
    if(stopped.compareAndSet(false,true)){
      eventThread.interrupt()
      var onStoppedCalled = false
      try{
        eventThread.join()
        onStoppedCalled = true
        onStop()
      } catch {
        case ie: InterruptedException =>
          Thread.currentThread().interrupt()
          if(!onStoppedCalled){
            onStop()
          }
      }
    }
  }


  def post(event:E): Unit={
    eventQueue.put(event)
  }

  def isActive:Boolean = eventThread.isAlive
  
  protected def onStart() : Unit ={}
  protected def onStop() : Unit = {}

  private[khaos] def onClear(): Unit ={
    eventQueue.clear()
  }

}
